Fox's Git Mirrors
Commit b06ec00b104f1e043aa38fa23dccd9d2c9010d53
Parents : 3bfd506
Author : Ivan <e318cbc04468bd574db2b4523dddd710>
Signature : T66BB85Valid, signed by author
Date : 2026-08-13T09:30:28-05:00
fix: match Python Channel RX order, send window/MDU, and HMU segment bounds
Changes
12 files changed, 317 insertions(+), 35 deletions(-)
Diff
diff --git a/pkg/buffer/buffer.go b/pkg/buffer/buffer.go
index d46c728e..b50ce68e 100644
--- a/pkg/buffer/buffer.go
+++ b/pkg/buffer/buffer.go
@@ -6,6 +6,7 @@ package buffer
import (
"bufio"
"bytes"
+ "context"
"encoding/binary"
"io"
"sync"
@@ -187,6 +188,9 @@ func (w *RawChannelWriter) Write(p []byte) (n int, err error) {
processed = len(p)
}
+ if err := w.channel.WaitReady(context.Background()); err != nil {
+ return 0, err
+ }
if err := w.channel.Send(msg); err != nil {
return 0, err
}
diff --git a/pkg/buffer/buffer_test.go b/pkg/buffer/buffer_test.go
index b417bb0c..78469e6d 100644
--- a/pkg/buffer/buffer_test.go
+++ b/pkg/buffer/buffer_test.go
@@ -223,8 +223,12 @@ func (m *mockLink) GetLinkID() []byte { retu
func (m *mockLink) Send(data []byte) any { return &packet.Packet{Raw: data} }
func (m *mockLink) Resend(p any) error { return nil }
func (m *mockLink) SetPacketTimeout(p any, cb func(any), t time.Duration) {}
-func (m *mockLink) SetPacketDelivered(p any, cb func(any)) {}
-func (m *mockLink) HandleInbound(pkt *packet.Packet) error { return nil }
+func (m *mockLink) SetPacketDelivered(p any, cb func(any)) {
+ if cb != nil {
+ cb(p)
+ }
+}
+func (m *mockLink) HandleInbound(pkt *packet.Packet) error { return nil }
func (m *mockLink) ValidateLinkProof(pkt *packet.Packet, networkIface common.NetworkInterface) error {
return nil
}
diff --git a/pkg/buffer/regression_test.go b/pkg/buffer/regression_test.go
index e8f5d20f..4219ead7 100644
--- a/pkg/buffer/regression_test.go
+++ b/pkg/buffer/regression_test.go
@@ -31,8 +31,12 @@ func (m *captureLink) Send(data []byte) any {
}
func (m *captureLink) Resend(p any) error { return nil }
func (m *captureLink) SetPacketTimeout(p any, cb func(any), t time.Duration) {}
-func (m *captureLink) SetPacketDelivered(p any, cb func(any)) {}
-func (m *captureLink) HandleInbound(pkt *packet.Packet) error { return nil }
+func (m *captureLink) SetPacketDelivered(p any, cb func(any)) {
+ if cb != nil {
+ cb(p)
+ }
+}
+func (m *captureLink) HandleInbound(pkt *packet.Packet) error { return nil }
func (m *captureLink) ValidateLinkProof(pkt *packet.Packet, networkIface common.NetworkInterface) error {
return nil
}
diff --git a/pkg/channel/channel.go b/pkg/channel/channel.go
index 3ddebdaa..8f249aba 100644
--- a/pkg/channel/channel.go
+++ b/pkg/channel/channel.go
@@ -17,9 +17,14 @@ import (
"quad4/reticulum-go/pkg/transport"
)
-// ErrLinkNotReady is returned when a send is attempted on a non-ready outlet.
+// ErrLinkNotReady is returned when a send is attempted on a non-ready outlet
+// or when the TX window is full, matching Python ChannelException ME_LINK_NOT_READY.
var ErrLinkNotReady = errors.New("link not ready")
+// ErrTooBig is returned when the packed envelope exceeds the outlet MDU,
+// matching Python ChannelException ME_TOO_BIG.
+var ErrTooBig = errors.New("channel message too big")
+
// SystemMessageTypeMin is the lower bound for system-reserved MSGTYPE values.
// Matches Python RNS Channel (MSGTYPE >= 0xf000).
const SystemMessageTypeMin uint16 = 0xf000
@@ -56,6 +61,7 @@ type Channel struct {
sendMu sync.Mutex
mutex sync.RWMutex
txRing []*Envelope
+ rxRing []rxEnvelope
window int
windowMax int
windowMin int
@@ -63,12 +69,18 @@ type Channel struct {
fastRateRounds int
mediumRateRounds int
nextSequence uint16
+ nextRxSequence uint16
maxTries int
messageHandlers []messageHandlerEntry
nextHandlerID int
factories map[uint16]MessageConstructor
}
+type rxEnvelope struct {
+ sequence uint16
+ message MessageBase
+}
+
type messageHandlerEntry struct {
id int
handler func(MessageBase) bool
@@ -161,11 +173,13 @@ func packEnvelope(msgType, sequence uint16, body []byte) ([]byte, error) {
// Send transmits a message over the channel.
// Sequence allocation and tx-ring emplace happen only after a successful
// outlet send so a failing link cannot leave ghost envelopes or sequence holes.
+// A full TX window or packed envelope larger than the outlet MDU is refused,
+// matching Python Channel.send.
func (c *Channel) Send(msg MessageBase) error {
c.sendMu.Lock()
defer c.sendMu.Unlock()
- if !outletReady(c.link.GetStatus()) {
+ if !c.IsReadyToSend() {
return ErrLinkNotReady
}
@@ -176,16 +190,23 @@ func (c *Channel) Send(msg MessageBase) error {
c.mutex.Lock()
reserved := c.nextSequence
- c.nextSequence = uint16((uint32(reserved) + 1) % SeqModulus)
c.mutex.Unlock()
raw, err := packEnvelope(msg.GetType(), reserved, body)
if err != nil {
- c.mutex.Lock()
- c.nextSequence = reserved
- c.mutex.Unlock()
return err
}
+ if len(raw) > c.outletMDU() {
+ return ErrTooBig
+ }
+
+ c.mutex.Lock()
+ if c.nextSequence != reserved {
+ c.mutex.Unlock()
+ return ErrLinkNotReady
+ }
+ c.nextSequence = uint16((uint32(reserved) + 1) % SeqModulus)
+ c.mutex.Unlock()
packet := c.link.Send(raw)
if !packetTransmitted(packet) {
@@ -305,7 +326,8 @@ func (c *Channel) RemoveMessageHandler(id int) {
}
// HandleInbound processes an inbound channel packet and dispatches to registered handlers.
-// Registered factories unpack into typed messages. Unknown types become GenericMessage.
+// Sequences are buffered on the RX ring and delivered in order, duplicates are
+// dropped, matching Python Channel._receive.
func (c *Channel) HandleInbound(data []byte) error {
if len(data) < ChannelHeaderSize {
return errors.New("channel packet too short")
@@ -319,14 +341,16 @@ func (c *Channel) HandleInbound(data []byte) error {
return errors.New("channel packet incomplete")
}
- msgData := make([]byte, length)
- copy(msgData, data[ChannelHeaderSize:ChannelHeaderSize+int(length)])
-
c.mutex.RLock()
+ stale := staleRXSequence(sequence, c.nextRxSequence)
ctor := c.factories[msgType]
- handlers := make([]messageHandlerEntry, len(c.messageHandlers))
- copy(handlers, c.messageHandlers)
c.mutex.RUnlock()
+ if stale {
+ return nil
+ }
+
+ msgData := make([]byte, length)
+ copy(msgData, data[ChannelHeaderSize:ChannelHeaderSize+int(length)])
var msg MessageBase
if ctor != nil {
@@ -342,9 +366,23 @@ func (c *Channel) HandleInbound(data []byte) error {
}
}
- for _, entry := range handlers {
- if entry.handler != nil {
- if entry.handler(msg) {
+ c.mutex.Lock()
+ if staleRXSequence(sequence, c.nextRxSequence) {
+ c.mutex.Unlock()
+ return nil
+ }
+ if !c.emplaceRXLocked(rxEnvelope{sequence: sequence, message: msg}) {
+ c.mutex.Unlock()
+ return nil
+ }
+ delivered := c.drainRXLocked()
+ handlers := make([]messageHandlerEntry, len(c.messageHandlers))
+ copy(handlers, c.messageHandlers)
+ c.mutex.Unlock()
+
+ for _, m := range delivered {
+ for _, entry := range handlers {
+ if entry.handler != nil && entry.handler(m) {
break
}
}
@@ -353,6 +391,56 @@ func (c *Channel) HandleInbound(data []byte) error {
return nil
}
+// staleRXSequence reports whether seq is behind nextRx and outside the wrap
+// window, matching Python Channel._receive WINDOW_MAX overflow logic.
+func staleRXSequence(seq, next uint16) bool {
+ if seq >= next {
+ return false
+ }
+ windowOverflow := uint16((uint32(next) + uint32(WindowMax)) % SeqModulus)
+ if windowOverflow < next {
+ return seq > windowOverflow
+ }
+ return true
+}
+
+func (c *Channel) emplaceRXLocked(env rxEnvelope) bool {
+ for i, existing := range c.rxRing {
+ if env.sequence == existing.sequence {
+ return false
+ }
+ dist := int32(c.nextRxSequence) - int32(env.sequence)
+ if env.sequence < existing.sequence && dist <= int32(SeqMax)/2 {
+ c.rxRing = append(c.rxRing, rxEnvelope{})
+ copy(c.rxRing[i+1:], c.rxRing[i:])
+ c.rxRing[i] = env
+ return true
+ }
+ }
+ c.rxRing = append(c.rxRing, env)
+ return true
+}
+
+func (c *Channel) drainRXLocked() []MessageBase {
+ var out []MessageBase
+ for {
+ found := -1
+ for i, env := range c.rxRing {
+ if env.sequence == c.nextRxSequence {
+ found = i
+ break
+ }
+ }
+ if found < 0 {
+ return out
+ }
+ env := c.rxRing[found]
+ c.rxRing = append(c.rxRing[:found], c.rxRing[found+1:]...)
+ out = append(out, env.message)
+ c.nextRxSequence = uint16((uint32(c.nextRxSequence) + 1) % SeqModulus)
+ }
+}
+
// GenericMessage is a default message implementation with type, data, and sequence.
type GenericMessage struct {
Type uint16
@@ -411,16 +499,22 @@ type mduOutlet interface {
GetMDU() int
}
-// MDU is bytes available for a channel message body, matching Python
-// Channel.mdu (outlet MDU minus 6-byte envelope header).
-func (c *Channel) MDU() int {
+func (c *Channel) outletMDU() int {
mdu := DefaultOutletMDU
- if g, ok := c.link.(mduOutlet); ok {
- if n := g.GetMDU(); n > 0 {
- mdu = n
+ if c.link != nil {
+ if g, ok := c.link.(mduOutlet); ok {
+ if n := g.GetMDU(); n > 0 {
+ mdu = n
+ }
}
}
- mdu -= ChannelHeaderSize
+ return mdu
+}
+
+// MDU is bytes available for a channel message body, matching Python
+// Channel.mdu (outlet MDU minus 6-byte envelope header).
+func (c *Channel) MDU() int {
+ mdu := c.outletMDU() - ChannelHeaderSize
if mdu > 0xFFFF {
mdu = 0xFFFF
}
@@ -522,5 +616,6 @@ func (c *Channel) Close() error {
releaseEnvelope(env)
}
c.txRing = nil
+ c.rxRing = nil
return nil
}
diff --git a/pkg/channel/channel_test.go b/pkg/channel/channel_test.go
index dc17c574..4b5e7019 100644
--- a/pkg/channel/channel_test.go
+++ b/pkg/channel/channel_test.go
@@ -119,7 +119,7 @@ func TestHandleInboundTypedFactory(t *testing.T) {
return true
})
- raw, err := packEnvelope(1, 7, []byte("abcd"))
+ raw, err := packEnvelope(1, 0, []byte("abcd"))
if err != nil {
t.Fatal(err)
}
@@ -142,7 +142,7 @@ func TestHandleInbound(t *testing.T) {
})
// Packet format: [type 2][seq 2][len 2][data]
- data := []byte{0, 1, 0, 1, 0, 4, 't', 'e', 's', 't'}
+ data := []byte{0, 1, 0, 0, 0, 4, 't', 'e', 's', 't'}
err := c.HandleInbound(data)
if err != nil {
t.Fatalf("HandleInbound failed: %v", err)
@@ -190,6 +190,7 @@ type scaleMockLink struct {
}
func (m *scaleMockLink) GetStatus() byte { return m.status }
+func (m *scaleMockLink) GetMDU() int { return DefaultOutletMDU }
func (m *scaleMockLink) GetRTT() float64 { return 0.1 }
func (m *scaleMockLink) RTT() float64 { return 0.1 }
func (m *scaleMockLink) GetLinkID() []byte { return []byte("mocklink") }
@@ -243,14 +244,14 @@ func BenchmarkChannelSendScale(b *testing.B) {
b.ReportAllocs()
for i := 0; i < b.N; i++ {
- _ = ch.Send(msg)
- if i%100 == 0 {
+ if err := ch.Send(msg); err != nil {
ch.mutex.Lock()
for _, env := range ch.txRing {
releaseEnvelope(env)
}
ch.txRing = nil
ch.mutex.Unlock()
+ _ = ch.Send(msg)
}
}
ch.mutex.Lock()
diff --git a/pkg/channel/fuzz_test.go b/pkg/channel/fuzz_test.go
index 37fd914b..d1e34304 100644
--- a/pkg/channel/fuzz_test.go
+++ b/pkg/channel/fuzz_test.go
@@ -45,6 +45,13 @@ func FuzzHandleInboundEnvelopeExploratory(f *testing.F) {
if len(data) < ChannelHeaderSize+int(length) {
t.Fatal("HandleInbound succeeded when declared length exceeds buffer")
}
+ seq := binary.BigEndian.Uint16(data[2:4])
+ if seq != 0 {
+ if got != nil {
+ t.Fatal("nonzero first sequence must not dispatch")
+ }
+ return
+ }
gm, ok := got.(*GenericMessage)
if !ok {
return
@@ -85,6 +92,12 @@ func FuzzPackHandleInboundRoundTrip(f *testing.F) {
if err := ch.HandleInbound(raw); err != nil {
t.Fatalf("HandleInbound: %v", err)
}
+ if seq != 0 {
+ if got != nil {
+ t.Fatal("nonzero first sequence must not dispatch")
+ }
+ return
+ }
if got == nil {
t.Fatal("handler did not receive GenericMessage")
}
diff --git a/pkg/channel/oracle_rx_sequence_test.go b/pkg/channel/oracle_rx_sequence_test.go
new file mode 100644
index 00000000..36896f64
--- /dev/null
+++ b/pkg/channel/oracle_rx_sequence_test.go
@@ -0,0 +1,113 @@
+// SPDX-License-Identifier: Apache-2.0
+// Copyright (c) 2024-2026 Quad4.io
+
+package channel
+
+import (
+ "errors"
+ "testing"
+
+ "quad4/reticulum-go/pkg/transport"
+)
+
+func TestHandleInboundReordersAndDropsDuplicates(t *testing.T) {
+ link := &mockLink{status: transport.StatusActive}
+ c := NewChannel(link)
+ defer func() { _ = c.Close() }()
+
+ if err := c.RegisterMessageType(1, func() MessageBase { return &testMessage{} }); err != nil {
+ t.Fatal(err)
+ }
+
+ var got []string
+ c.AddMessageHandler(func(msg MessageBase) bool {
+ if m, ok := msg.(*testMessage); ok {
+ got = append(got, string(m.data))
+ }
+ return true
+ })
+
+ raw1, err := packEnvelope(1, 1, []byte("B"))
+ if err != nil {
+ t.Fatal(err)
+ }
+ raw0, err := packEnvelope(1, 0, []byte("A"))
+ if err != nil {
+ t.Fatal(err)
+ }
+
+ if err := c.HandleInbound(raw1); err != nil {
+ t.Fatal(err)
+ }
+ if len(got) != 0 {
+ t.Fatalf("seq 1 before 0 dispatched %v", got)
+ }
+ if err := c.HandleInbound(raw0); err != nil {
+ t.Fatal(err)
+ }
+ if len(got) != 2 || got[0] != "A" || got[1] != "B" {
+ t.Fatalf("order=%v want [A B]", got)
+ }
+
+ if err := c.HandleInbound(raw0); err != nil {
+ t.Fatal(err)
+ }
+ if len(got) != 2 {
+ t.Fatalf("duplicate seq 0 dispatched, got %v", got)
+ }
+}
+
+func TestSendRefusesFullWindow(t *testing.T) {
+ link := &mockLink{status: transport.StatusActive}
+ c := NewChannel(link)
+ defer func() { _ = c.Close() }()
+
+ for i := 0; i < WindowInitial; i++ {
+ if err := c.Send(&testMessage{data: []byte{byte(i)}}); err != nil {
+ t.Fatalf("setup send %d: %v", i, err)
+ }
+ }
+ if c.IsReadyToSend() {
+ t.Fatal("window should be full")
+ }
+ err := c.Send(&testMessage{data: []byte("overflow")})
+ if !errors.Is(err, ErrLinkNotReady) {
+ t.Fatalf("Send: got %v want ErrLinkNotReady", err)
+ }
+ if c.TxRingLen() != WindowInitial {
+ t.Fatalf("tx ring len=%d want %d", c.TxRingLen(), WindowInitial)
+ }
+}
+
+func TestSendRefusesOversizeEnvelope(t *testing.T) {
+ link := &mockLink{status: transport.StatusActive}
+ c := NewChannel(link)
+ defer func() { _ = c.Close() }()
+
+ body := make([]byte, c.MDU()+64)
+ err := c.Send(&testMessage{data: body})
+ if !errors.Is(err, ErrTooBig) {
+ t.Fatalf("Send: got %v want ErrTooBig", err)
+ }
+ if c.TxRingLen() != 0 {
+ t.Fatalf("tx ring len=%d want 0", c.TxRingLen())
+ }
+ if c.NextSequence() != 0 {
+ t.Fatalf("sequence consumed on oversized send: %d", c.NextSequence())
+ }
+}
+
+func TestStaleRXSequenceWrapWindow(t *testing.T) {
+ if staleRXSequence(5, 10) != true {
+ t.Fatal("seq 5 must be stale when next is 10")
+ }
+ if staleRXSequence(10, 10) != false {
+ t.Fatal("current next sequence is not stale")
+ }
+ if staleRXSequence(0, 65530) != false {
+ t.Fatal("seq 0 after wrap must be accepted while next is 65530")
+ }
+ if staleRXSequence(100, 65530) != true {
+ t.Fatal("seq 100 must be stale when next is 65530")
+ }
+}
diff --git a/pkg/channel/regression_test.go b/pkg/channel/regression_test.go
index 7d3ca4d0..880f9b34 100644
--- a/pkg/channel/regression_test.go
+++ b/pkg/channel/regression_test.go
@@ -73,7 +73,7 @@ func TestRegression_HandleInboundUsesRegisteredFactory(t *testing.T) {
return true
})
- raw, err := packEnvelope(1, 3, []byte("typed"))
+ raw, err := packEnvelope(1, 0, []byte("typed"))
if err != nil {
t.Fatal(err)
}
diff --git a/pkg/link/hmu_negative_segment_oracle_test.go b/pkg/link/hmu_negative_segment_oracle_test.go
index 5565d467..02e14095 100644
--- a/pkg/link/hmu_negative_segment_oracle_test.go
+++ b/pkg/link/hmu_negative_segment_oracle_test.go
@@ -50,4 +50,14 @@ func TestOracleHMUOversizeSegmentDoesNotPanic(t *testing.T) {
if added != 0 {
t.Fatalf("expected 0 entries applied for oversize segment, got %d", added)
}
+
+ added = rx.applyHashmapSegment(1<<61, hashmapBytes)
+ if added != 0 {
+ t.Fatalf("expected 0 entries applied for overflowing segment, got %d", added)
+ }
+ for i, mh := range rx.mapHashes {
+ if mh != nil {
+ t.Fatalf("mapHashes[%d] was populated by an overflowing-segment HMU", i)
+ }
+ }
}
diff --git a/pkg/link/incoming_resource.go b/pkg/link/incoming_resource.go
index 27748810..82622fff 100644
--- a/pkg/link/incoming_resource.go
+++ b/pkg/link/incoming_resource.go
@@ -129,9 +129,8 @@ type incomingResourceAsm struct {
func (rx *incomingResourceAsm) applyHashmapSegment(segment int, hashmapBytes []byte) int {
// segment is parsed straight off the wire in handleResourceHashmapUpdate
// (wireInt(update[0])) and can be any attacker-chosen int, including
- // negative. Reject it here rather than trusting callers, since a
- // negative segment*segLen would otherwise produce a negative slice
- // index below and panic the process (remote DoS).
+ // negative or so large that segment*segLen overflows int. Reject
+ // those here. A wrapping multiply would otherwise write hashmap[0].
if segment < 0 {
return 0
}
@@ -139,10 +138,17 @@ func (rx *incomingResourceAsm) applyHashmapSegment(segment int, hashmapBytes []b
if segLen <= 0 {
segLen = 1
}
+ if segment > math.MaxInt/segLen {
+ return 0
+ }
+ base := segment * segLen
added := 0
hashes := len(hashmapBytes) / resource.MapHashLen
for i := range hashes {
- idx := i + segment*segLen
+ if i > math.MaxInt-base {
+ return added
+ }
+ idx := base + i
if idx < 0 || idx >= rx.totalParts {
return added
}
diff --git a/pkg/link/oracle_hmu_wrap_test.go b/pkg/link/oracle_hmu_wrap_test.go
new file mode 100644
index 00000000..6d747591
--- /dev/null
+++ b/pkg/link/oracle_hmu_wrap_test.go
@@ -0,0 +1,32 @@
+// SPDX-License-Identifier: Apache-2.0
+// Copyright (c) 2024-2026 Quad4.io
+
+package link
+
+import (
+ "bytes"
+ "testing"
+
+ "quad4/reticulum-go/pkg/resource"
+)
+
+func TestHMUSegmentIndexWrapDoesNotOverwriteSlotZero(t *testing.T) {
+ rx := &incomingResourceAsm{
+ hashmapSegLen: 8,
+ totalParts: 16,
+ partSlots: make([][]byte, 16),
+ mapHashes: make([][]byte, 16),
+ }
+
+ marker := bytes.Repeat([]byte{0xAA}, resource.MapHashLen)
+ hashmapBytes := make([]byte, resource.MapHashLen*4)
+ copy(hashmapBytes, marker)
+
+ added := rx.applyHashmapSegment(1<<61, hashmapBytes)
+ if added != 0 {
+ t.Fatalf("wrapped segment applied %d entries", added)
+ }
+ if rx.mapHashes[0] != nil {
+ t.Fatalf("slot 0 overwritten by wrapping HMU segment: %x", rx.mapHashes[0])
+ }
+}
diff --git a/reticulum-go.rsm b/reticulum-go.rsm
index 21eaa44b..7b87ac73 100644
Binary files a/reticulum-go.rsm and b/reticulum-go.rsm differ
Served by rngit 1.5.0 - Generated in 0.14s